package com.reliable.service.serviceimpl;

import com.alibaba.dashscope.aigc.generation.Generation;
import com.alibaba.dashscope.aigc.generation.GenerationParam;
import com.alibaba.dashscope.aigc.generation.GenerationResult;
import com.alibaba.dashscope.common.Message;
import com.alibaba.dashscope.common.Role;
import com.alibaba.excel.EasyExcel;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.reliable.api.dto.chatmessage.*;

import com.reliable.api.dto.ReliableTaskStatDTO;
import com.reliable.api.dto.excel.ChatMessageExcelDTO;
import com.reliable.api.dto.largescreen.TotalOverviewData;
import com.reliable.api.dto.user.ReliableUserLite;
import com.reliable.common.annotation.DataSource;
import com.reliable.common.enums.DataSourceType;
import com.reliable.common.utils.StringUtils;
import com.reliable.framework.datasource.DynamicDataSourceContextHolder;
import com.reliable.framework.utils.JwtUtils;
import com.reliable.service.domain.*;
import com.reliable.service.enums.UploadRecordStatus;
import com.reliable.service.excel.ChatMessageRow;
import com.reliable.service.listener.ChatMessageImportListener;
import com.reliable.service.mapper.ChatMessageMapper;
import com.reliable.service.service.*;

import com.reliable.service.utils.DataSourceUtil;

import com.reliable.service.utils.IdEncryptUtils;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import java.io.*;
import java.math.BigDecimal;
import java.sql.*;
import java.time.LocalDateTime;
import java.time.OffsetDateTime;
import java.time.YearMonth;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

@Service
@Slf4j
public class ChatMessageServiceImpl implements ChatMessageService {

    @Autowired
    private ChatMessageMapper chatMessageMapper;

    @Autowired
    private javax.sql.DataSource dataSource;


    @Autowired
    private IReliabUploadRecordService uploadRecordService;

    @Autowired
    private IReliableAnalyzedSessionService analyzedSessionService;

    @Autowired
    private IReliableTransactionRecordService transactionRecordService;

    @Autowired
    private IReliableTemplateAnalysisTaskService templateAnalysisTaskService;

    @Autowired
    @Qualifier("pgJdbcTemplate")
    private JdbcTemplate jdbcTemplate;

    // 或者更推荐：注入你已有的 Prompt Service
    @Autowired
    private IAiAnalysisPromptService aiPromptService; // ← 推荐方式

    // 添加一个用于AI调用的线程池
    private final ExecutorService aiExecutor = Executors.newFixedThreadPool(5);
    // 👇 新增线程池（用于并行处理会话）
    private final ExecutorService sessionExecutor = Executors.newFixedThreadPool(
            Math.min(10, Runtime.getRuntime().availableProcessors())
    );


    /**
     * 根据任务ID获取任务统计信息
     *
     * @param taskId 任务ID
     * @return 任务统计信息
     */
    public ReliableTaskStatDTO getTaskStatistics(Long taskId) {
        DataSourceUtil.setDataSourceType(DataSourceType.PG.name());
        try {
            String sql = "SELECT " +
                    "COUNT(DISTINCT session_id) AS sessionCount, " +
                    "COUNT(DISTINCT visitor_id) AS entidCount, " +
                    "COUNT(*) AS messageCount, " +
                    "COUNT(DISTINCT visitor_ip) AS ipCount, " +
                    "COUNT(DISTINCT referrer_url) FILTER (WHERE referrer_url IS NOT NULL AND referrer_url != '') AS referrerUrlCount, " +
                    "COUNT(DISTINCT page_url) FILTER (WHERE page_url IS NOT NULL AND page_url != '') AS pageUrlCount, " +
                    "MIN(created_at) AS earliestTime, " +
                    "MAX(created_at) AS latestTime " +
                    "FROM public.chat_messages " +
                    "WHERE task_id = ?";

            List<ReliableTaskStatDTO> result = jdbcTemplate.query(sql, (rs, rowNum) -> {
                ReliableTaskStatDTO statDTO = new ReliableTaskStatDTO();
                statDTO.setTaskId(taskId);
                statDTO.setSessionCount(rs.getLong("sessionCount"));
                statDTO.setEntCount(rs.getLong("entidCount"));
                statDTO.setMessageCount(rs.getLong("messageCount"));
                statDTO.setIpCount(rs.getLong("ipCount"));
                statDTO.setVisitorIpCount(rs.getLong("ipCount")); // visitorIpCount与ipCount相同
                statDTO.setReferrerUrlCount(rs.getLong("referrerUrlCount"));
                statDTO.setPageUrlCount(rs.getLong("pageUrlCount"));
                statDTO.setEarliestTime(rs.getTimestamp("earliestTime") != null ? rs.getTimestamp("earliestTime").toLocalDateTime() : null);
                statDTO.setLatestTime(rs.getTimestamp("latestTime") != null ? rs.getTimestamp("latestTime").toLocalDateTime() : null);
                return statDTO;
            }, taskId);

            if (!result.isEmpty()) {
                return result.get(0);
            }

            // 返回默认值
            ReliableTaskStatDTO defaultStat = new ReliableTaskStatDTO();
            defaultStat.setTaskId(taskId);
            defaultStat.setSessionCount(0L);
            defaultStat.setEntCount(0L);
            defaultStat.setMessageCount(0L);
            defaultStat.setIpCount(0L);
            defaultStat.setVisitorIpCount(0L);
            defaultStat.setReferrerUrlCount(0L);
            defaultStat.setPageUrlCount(0L);
            return defaultStat;
        } finally {
            DataSourceUtil.clearDataSourceType();
        }
    }

    /**
     * 根据任务ID集合获取任务统计信息
     *
     * @param taskIds 任务ID集合
     * @return 任务统计信息列表
     */
    public List<ReliableTaskStatDTO> getTaskStatisticsBatch(List<Long> taskIds) {
        if (taskIds == null || taskIds.isEmpty()) {
            return new ArrayList<>();
        }

        DataSourceUtil.setDataSourceType(DataSourceType.PG.name());
        try {
            // 构建SQL中的占位符
            String placeholders = taskIds.stream().map(id -> "?").collect(Collectors.joining(","));
            String sql = "SELECT " +
                    "task_id AS taskId, " +
                    "COUNT(DISTINCT session_id) AS sessionCount, " +
                    "COUNT(DISTINCT visitor_id) AS entidCount, " +
                    "COUNT(*) AS messageCount, " +
                    "COUNT(DISTINCT visitor_ip) AS ipCount, " +
                    "COUNT(DISTINCT referrer_url) FILTER (WHERE referrer_url IS NOT NULL AND referrer_url != '') AS referrerUrlCount, " +
                    "COUNT(DISTINCT page_url) FILTER (WHERE page_url IS NOT NULL AND page_url != '') AS pageUrlCount, " +
                    "MIN(created_at) AS earliestTime, " +
                    "MAX(created_at) AS latestTime " +
                    "FROM public.chat_messages " +
                    "WHERE task_id IN (" + placeholders + ") " +
                    "GROUP BY task_id";

            return jdbcTemplate.query(sql, (rs, rowNum) -> {
                ReliableTaskStatDTO statDTO = new ReliableTaskStatDTO();
                statDTO.setTaskId(rs.getLong("taskId"));
                statDTO.setSessionCount(rs.getLong("sessionCount"));
                statDTO.setEntCount(rs.getLong("entidCount"));
                statDTO.setMessageCount(rs.getLong("messageCount"));
                statDTO.setIpCount(rs.getLong("ipCount"));
                statDTO.setVisitorIpCount(rs.getLong("ipCount")); // visitorIpCount与ipCount相同
                statDTO.setReferrerUrlCount(rs.getLong("referrerUrlCount"));
                statDTO.setPageUrlCount(rs.getLong("pageUrlCount"));
                statDTO.setEarliestTime(rs.getTimestamp("earliestTime") != null ? rs.getTimestamp("earliestTime").toLocalDateTime() : null);
                statDTO.setLatestTime(rs.getTimestamp("latestTime") != null ? rs.getTimestamp("latestTime").toLocalDateTime() : null);
                return statDTO;
            }, taskIds.toArray());
        } finally {
            DataSourceUtil.clearDataSourceType();
        }
    }

    /**
     * 安全获取Long值
     */
    private Long getLongValue(Object obj) {
        if (obj == null) return 0L;
        if (obj instanceof Long) return (Long) obj;
        if (obj instanceof Number) return ((Number) obj).longValue();
        try {
            return Long.parseLong(obj.toString());
        } catch (NumberFormatException e) {
            return 0L;
        }
    }

    /**
     * 安全获取LocalDateTime值
     */
    private LocalDateTime getLocalDateTimeValue(Object obj) {
        if (obj == null) return null;
        if (obj instanceof LocalDateTime) return (LocalDateTime) obj;
        if (obj instanceof Timestamp) return ((Timestamp) obj).toLocalDateTime();
        return null;
    }

    @Override
    @DataSource(value = DataSourceType.PG)
    public void save(ChatMessage message) {
        System.out.println("当前数据源: " + DataSourceUtil.getCurrentDataSource());
        // 确保分区存在
        createPartitionsIfNeeded(message.getCreatedAt(), message.getCreatedAt());
        chatMessageMapper.insert(message);
    }

    @Override
    @DataSource(value = DataSourceType.PG)
    public java.util.List<ChatMessage> findBySessionId(String sessionId) {
        return chatMessageMapper.selectList(
                new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper<ChatMessage>()
                        .eq(ChatMessage::getSessionId, sessionId)
                        .orderByAsc(ChatMessage::getCreatedAt)
        );
    }

    @Override
    @DataSource(value = DataSourceType.PG)
    public java.util.List<ChatMessage> findByTaskId(Long taskId) {
        return chatMessageMapper.selectList(
                new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper<ChatMessage>()
                        .eq(ChatMessage::getTaskId, taskId)
                        .orderByAsc(ChatMessage::getCreatedAt)
        );
    }
    // ========== 核心方法：Excel 导入 ==========


    @Override
    @DataSource(value = DataSourceType.MASTER)
    public void importExcel(MultipartFile file, String taskId) throws Exception {
        ReliableUserLite currentUser = JwtUtils.getCurrentUser();
        if (StringUtils.isEmpty(taskId))
        {
            log.error("任务ID不能为空");
            return;
        }
        Long decrypt = IdEncryptUtils.decrypt(taskId);
        log.info("开始处理Excel文件，任务ID: {}", decrypt);
        ReliabUploadRecord uploadRecord = new ReliabUploadRecord();
        uploadRecord.setTaskId(decrypt);
        uploadRecord.setFileName(file.getOriginalFilename());
        uploadRecord.setStatus(UploadRecordStatus.PROCESSING.getCode());
        uploadRecordService.save(uploadRecord);

        LambdaQueryWrapper<ReliableTemplateAnalysisTask> reliableTask = new LambdaQueryWrapper<>();
        reliableTask.eq(ReliableTemplateAnalysisTask::getTemplateId, decrypt);

        ReliableTemplateAnalysisTask task = templateAnalysisTaskService.getOne(reliableTask);
        task.setUploadRecordId(uploadRecord.getId());
        templateAnalysisTaskService.updateById(task);

        importExcelAsync(file, decrypt, uploadRecord.getId()).get();
    }

    @Async
    public CompletableFuture<Void> importExcelAsync(MultipartFile file, Long taskId, Long uploadId) {
        long startTime = System.currentTimeMillis();
        log.info("开始处理Excel文件，任务ID: {}, 上传ID: {}", taskId, uploadId);

        log.info("🔧 异步线程初始数据源上下文: {}", DynamicDataSourceContextHolder.getDataSourceType());
        log.info("🔧 注入的 dataSource 类型: {}", dataSource.getClass().getSimpleName());

        File tempFile = null;
        ChatMessageImportListener listener = null;
        try {
            tempFile = File.createTempFile("excel_import_", ".xlsx");
            file.transferTo(tempFile);

            // ✅ 整个 Excel 解析 + 插入 + 分区创建 阶段都使用 PG
            DynamicDataSourceContextHolder.setDataSourceType(DataSourceType.PG.name());
            log.info("🔄 已切换到 PG 数据源，准备解析 Excel");

            listener = new ChatMessageImportListener(taskId, uploadId, this);
            EasyExcel.read(tempFile, ChatMessageExcelDTO.class, listener)
                    .sheet()
                    .doRead();

            log.info("✅ Excel 解析完成，成功行数: {}, 失败行数: {}",
                    listener.getSuccessRows(), listener.getFailedRows());

            // ✅ 切换回 MASTER (MySQL) 更新上传记录
            DynamicDataSourceContextHolder.setDataSourceType(DataSourceType.MASTER.name());
            log.info("🔄 已切换回 MASTER (MySQL) 数据源，准备更新上传记录");

            uploadRecordService.updateUploadRecordSuccess(
                    uploadId,
                    listener.getTotalRows(),
                    listener.getSuccessRows(),
                    listener.getFailedRows()
            );

            long endTime = System.currentTimeMillis();
            log.info("✅ Excel导入完成，总耗时: {}ms", (endTime - startTime));
            return CompletableFuture.completedFuture(null);

        } catch (Exception e) {
            log.error("❌ Excel导入过程中发生异常", e);
            try {
                DynamicDataSourceContextHolder.setDataSourceType(DataSourceType.MASTER.name());
                log.info("🔄 异常处理：已切换到 MASTER (MySQL) 数据源");
                String errorMsg = e.getMessage() != null ? e.getMessage() : "未知错误";
                uploadRecordService.updateUploadRecordFailed(uploadId, errorMsg);
            } catch (Exception ex) {
                log.error("⚠️ 更新上传记录失败（异常处理中）", ex);
            }
            throw new RuntimeException("Excel导入失败", e);
        } finally {
            DynamicDataSourceContextHolder.clearDataSourceType();
            if (tempFile != null && tempFile.exists()) {
                try {
                    tempFile.delete();
                } catch (Exception e) {
                    log.warn("临时文件删除失败: {}", tempFile.getAbsolutePath(), e);
                }
            }
        }
    }



    @Override
    public void batchInsertChatMessages(List<ChatMessageRow> rows) {
        if (rows == null || rows.isEmpty()) return;

        LocalDateTime minTime = rows.stream()
                .map(ChatMessageRow::getCreatedAt)
                .min(LocalDateTime::compareTo)
                .orElseThrow(() -> new IllegalArgumentException("createdAt 不能为空"));
        LocalDateTime maxTime = rows.stream()
                .map(ChatMessageRow::getCreatedAt)
                .max(LocalDateTime::compareTo)
                .orElseThrow(() -> new IllegalArgumentException("createdAt 不能为空"));

        // 此时上下文已是 PG（由 importExcelAsync 设置）
        createPartitionsIfNeeded(minTime, maxTime);

        Map<String, List<ChatMessageRow>> groupedRows = rows.stream()
                .collect(Collectors.groupingBy(row ->
                        row.getCreatedAt().format(DateTimeFormatter.ofPattern("yyyy_MM"))
                ));

        for (Map.Entry<String, List<ChatMessageRow>> entry : groupedRows.entrySet()) {
            insertIntoSinglePartition(entry.getKey(), entry.getValue());
        }
    }

    private void insertIntoSinglePartition(String partitionSuffix, List<ChatMessageRow> rows) {
        String targetTable = "chat_messages_" + partitionSuffix;
        String tempTable = "temp_import_" + System.currentTimeMillis() + "_" + Thread.currentThread().getId();
        tempTable = tempTable.replaceAll("[^a-zA-Z0-9_]", "_");

        // 不再手动 set/clear！上下文由外层保证
        Connection conn = null;
        try {
            conn = dataSource.getConnection();
            conn.setAutoCommit(false);

            try (Statement stmt = conn.createStatement()) {
                stmt.execute("CREATE TEMP TABLE " + tempTable + " (LIKE " + targetTable + " INCLUDING DEFAULTS INCLUDING COMMENTS)");
            }

            String insertSql = "INSERT INTO " + tempTable +
                    " (company_id, session_id, content, visitor_id, role, agent_account, visitor_ip, created_at, source_type, task_id, content_hash, referrer_url, page_url) " +
                    " VALUES (?, ?, ?, ?, ?, ?, ?::inet, ?, ?, ?, md5(?), ?, ?)";

            try (PreparedStatement ps = conn.prepareStatement(insertSql)) {
                for (ChatMessageRow row : rows) {
                    String content = row.getContent() != null ? row.getContent() : "";
                    ps.setObject(1, row.getCompanyId());
                    ps.setString(2, row.getSessionId());
                    ps.setString(3, content);
                    ps.setString(4, row.getVisitorId());
                    ps.setString(5, row.getRole());
                    ps.setString(6, row.getAgentAccount());
                    ps.setString(7, row.getVisitorIp());
                    ps.setTimestamp(8, java.sql.Timestamp.valueOf(row.getCreatedAt()));
                    ps.setString(9, row.getSourceType());
                    ps.setLong(10, row.getTaskId());
                    ps.setString(11, content);
                    ps.setString(12, row.getReferrerUrl());
                    ps.setString(13, row.getPageUrl());
                    ps.addBatch();
                }
                ps.executeBatch();
            }

            String mergeSql = "INSERT INTO " + targetTable +
                    " SELECT * FROM " + tempTable +
                    " ON CONFLICT (task_id, session_id, role, content_hash, created_at) DO NOTHING";

            try (Statement stmt = conn.createStatement()) {
                int inserted = stmt.executeUpdate(mergeSql);
                log.info("✅ 分区 {} 导入 {} 行", targetTable, inserted);
            }

            try (Statement stmt = conn.createStatement()) {
                stmt.execute("DROP TABLE IF EXISTS " + tempTable);
                log.info("✅ 已删除临时表: {}", tempTable);
            }

            conn.commit();
        } catch (Exception e) {
            if (conn != null) {
                try {
                    conn.rollback();
                    try (Statement stmt = conn.createStatement()) {
                        stmt.execute("DROP TABLE IF EXISTS " + tempTable);
                        log.info("✅ 回滚后已删除临时表: {}", tempTable);
                    }
                } catch (Exception rollbackEx) {
                    log.error("回滚或删除临时表失败", rollbackEx);
                }
            }
            throw new RuntimeException("批量插入失败", e);
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (Exception closeEx) {
                    log.error("关闭连接失败", closeEx);
                }
            }
            // 不再 clear！由 importExcelAsync 统一清理
        }
    }

    @Override
// @DataSource(value = DataSourceType.PG) ← 移除！
    public void createPartitionsIfNeeded(LocalDateTime minTime, LocalDateTime maxTime) {
        // 不再手动 set/clear！上下文由外层保证
        try {
            YearMonth start = YearMonth.from(minTime.withDayOfMonth(1));
            YearMonth end = YearMonth.from(maxTime.withDayOfMonth(1));
            Set<String> existingPartitions = getExistingPartitions();

            for (YearMonth current = start; !current.isAfter(end); current = current.plusMonths(1L)) {
                String partitionName = "chat_messages_" + current.format(DateTimeFormatter.ofPattern("yyyy_MM"));
                if (!existingPartitions.contains(partitionName)) {
                    String startMonth = current.atDay(1).format(DateTimeFormatter.ISO_LOCAL_DATE);
                    String endMonth = current.plusMonths(1L).atDay(1).format(DateTimeFormatter.ISO_LOCAL_DATE);
                    String sql = String.format(
                            "CREATE TABLE IF NOT EXISTS %s PARTITION OF chat_messages FOR VALUES FROM ('%s') TO ('%s')",
                            partitionName, startMonth, endMonth
                    );

                    try (Connection conn = dataSource.getConnection();
                         Statement stmt = conn.createStatement()) {
                        stmt.execute(sql);
                        log.info("✅ 创建分区: {}", partitionName);
                    } catch (Exception e) {
                        log.error("❌ 创建分区失败: {}", partitionName, e);
                        throw e; // 不要吞异常
                    }
                }
            }
        } catch (Exception e) {
            log.error("创建分区过程中发生异常", e);
            throw new RuntimeException("创建分区失败", e);
        }
    }

    private Set<String> getExistingPartitions() {
        // 不再手动 set/clear！
        Set<String> partitions = new HashSet<>();
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement()) {
            ResultSet rs = stmt.executeQuery("SELECT tablename FROM pg_tables WHERE tablename LIKE 'chat_messages_%'");
            while (rs.next()) {
                partitions.add(rs.getString("tablename"));
            }
        } catch (Exception e) {
            log.error("获取现有分区失败", e);
            throw new RuntimeException("获取分区列表失败", e);
        }
        return partitions;
    }

    @Override
// @DataSource(value = DataSourceType.PG) ← 移除！
    public void saveUploadSessionRef(Long uploadId, Set<String> sessionIds) throws Exception {
        log.info("📥 开始保存 upload_session_ref，uploadId={}, sessionIds数量={}", uploadId, sessionIds != null ? sessionIds.size() : 0);
        if (sessionIds == null || sessionIds.isEmpty()) {
            log.info("⚠️ sessionIds 为空，跳过保存 upload_session_ref");
            return;
        }

        // 不再手动 set/clear！由调用方（doAfterAllAnalysed）控制
        String sql = "INSERT INTO upload_session_ref (upload_id, session_id) VALUES (?, ?)";
        try (Connection conn = dataSource.getConnection();
             PreparedStatement ps = conn.prepareStatement(sql)) {

            for (String sessionId : sessionIds) {
                if (sessionId != null) {
                    ps.setLong(1, uploadId);
                    ps.setString(2, sessionId);
                    ps.addBatch();
                }
            }

            int[] results = ps.executeBatch();
            log.info("✅ 成功插入 {} 条 upload_session_ref 记录，uploadId={}", results.length, uploadId);
        } catch (Exception e) {
            log.error("❌ 保存 upload_session_ref 失败，uploadId={}", uploadId, e);
            throw e;
        }
    }


    /**
     * 分析所有会话（入口方法）- 优化版本，支持并发处理
     */
    public void analyzeAllChatSessions() {
        DataSourceUtil.setDataSourceType(DataSourceType.PG.name());
        String sessionSql = "SELECT DISTINCT session_id FROM chat_messages WHERE session_id IS NOT NULL ORDER BY session_id";
        List<String> sessionIds = jdbcTemplate.queryForList(sessionSql, String.class);
        System.out.println("总共找到 " + sessionIds.size() + " 个会话需要分析");
        DataSourceUtil.clearDataSourceType();

        // 并行处理所有会话

        // 等待全部完成
        CompletableFuture.allOf(sessionIds.stream()
                .map(sessionId -> CompletableFuture.runAsync(() -> processSingleSession(sessionId), sessionExecutor)).toArray(CompletableFuture[]::new)).join();
    }

    /**
     * 处理单个会话（使用 PG 数据源）- 异步非阻塞 AI 调用
     */
    @DataSource(DataSourceType.PG)
    public void processSingleSession(String sessionId) {
        try {
            // 1. 查询上次分析时间
            String checkSql = "SELECT last_analyzed_time FROM analyzed_sessions WHERE session_id = ?";
            List<Map<String, Object>> records = jdbcTemplate.queryForList(checkSql, sessionId);
            LocalDateTime lastAnalyzedTime = null;
            if (!records.isEmpty()) {
                Timestamp ts = (Timestamp) records.get(0).get("last_analyzed_time");
                if (ts != null) {
                    lastAnalyzedTime = ts.toLocalDateTime();
                }
            }

            // 2. 查询新增消息
            String messageSql;
            List<Map<String, Object>> messages;
            if (lastAnalyzedTime == null) {
                messageSql = "SELECT session_id, role, content, created_at, task_id FROM chat_messages WHERE session_id = ? ORDER BY created_at";
                messages = jdbcTemplate.queryForList(messageSql, sessionId);
            } else {
                messageSql = "SELECT session_id, role, content, created_at, task_id FROM chat_messages WHERE session_id = ? AND created_at > ? ORDER BY created_at";
                messages = jdbcTemplate.queryForList(messageSql, sessionId, Timestamp.valueOf(lastAnalyzedTime));
            }

            if (messages.isEmpty()) {
                System.out.println("会话 " + sessionId + " 无新增消息，跳过分析");
                return;
            }

            // 👇【关键优化】提前提取轻量级变量，避免 Lambda 捕获整个 messages 列表
            Long taskId = (Long) messages.get(0).get("task_id");
            int messageCount = messages.size();
            LocalDateTime newLastTime = ((Timestamp) messages.get(messages.size() - 1).get("created_at")).toLocalDateTime();

            // 3. 异步调用 AI，并链式保存结果
            List<Map<String, Object>> finalMessages = messages;
            CompletableFuture.supplyAsync(() -> callAIApi(finalMessages), aiExecutor)
                    .thenAccept(analysisResult -> {
                        try {
                            // 使用提前提取的轻量变量
                            saveAnalysisResultToMySQL(sessionId, messageCount, analysisResult, taskId);
                            recordAnalyzedSession(sessionId, messageCount, newLastTime);
                            System.out.println("✅ 会话 " + sessionId + " 分析完成，新增消息数: " + messageCount);
                        } catch (Exception e) {
                            System.err.println("保存结果失败: " + sessionId + ", 错误: " + e.getMessage());
                            e.printStackTrace();
                        }
                    })
                    .exceptionally(ex -> {
                        System.err.println("处理会话失败: " + sessionId + ", 错误: " + ex.getMessage());
                        return null;
                    });

            // 👇 方法结束，messages 可被 GC（因为 Lambda 不再引用它）
            // 可显式置空（非必须，但更保险）
            messages = null;

        } catch (Exception e) {
            System.err.println("初始化会话失败: " + sessionId + ", 错误: " + e.getMessage());
            e.printStackTrace();
        }
    }

    /**
     * 记录已分析会话（操作 MySQL，无需 @DataSource）
     * 优化版本 - 使用批量更新减少数据库交互
     */
    @DataSource(DataSourceType.MASTER)
    private void recordAnalyzedSession(String sessionId, int messageCount, LocalDateTime lastAnalyzedTime) {
        try {
            String updateSql = "UPDATE analyzed_sessions SET analyzed_at = ?, message_count = ?, last_analyzed_time = ? WHERE session_id = ?";
            int updated = jdbcTemplate.update(updateSql,
                    new Timestamp(System.currentTimeMillis()),
                    messageCount,
                    Timestamp.valueOf(lastAnalyzedTime),
                    sessionId
            );
            if (updated == 0) {
                String insertSql = "INSERT INTO analyzed_sessions (session_id, analyzed_at, message_count, last_analyzed_time) VALUES (?, ?, ?, ?)";
                jdbcTemplate.update(insertSql,
                        sessionId,
                        new Timestamp(System.currentTimeMillis()),
                        messageCount,
                        Timestamp.valueOf(lastAnalyzedTime)
                );
            }
        } catch (Exception e) {
            System.err.println("记录已分析会话失败: " + sessionId + ", 错误: " + e.getMessage());
        }
    }

    @DataSource(value = DataSourceType.PG)
    private String callAIApi(List<Map<String, Object>> messages) {
        try {
            if (messages == null || messages.isEmpty()) {
                return "{}";
            }

            //从 MySQL 动态获取 Prompt（使用你创建的 Service）
            AiAnalysisPrompt prompt = aiPromptService.getDefaultEnabledPrompt();
            if (prompt == null) {
                throw new RuntimeException("未配置默认AI分析提示词");
            }
            String systemPrompt = prompt.getPromptContent();

            // 构建用户消息内容（不变）
            StringBuilder userContent = new StringBuilder();
            for (Map<String, Object> message : messages) {
                String sessionId = (String) message.get("session_id");
                String role = (String) message.get("role");
                String content = (String) message.get("content");
                Timestamp createdAt = (Timestamp) message.get("created_at");

                userContent.append("\"").append(sessionId).append("\"\t")
                        .append("\"").append(role).append("\"\t")
                        .append("\"").append(content).append("\"\t")
                        .append("\"").append(createdAt.toString()).append("\"\n");
            }

            // 调用 AI（仅替换 content）
            Generation gen = new Generation();
            Message systemMsg = Message.builder()
                    .role(Role.SYSTEM.getValue())
                    .content(systemPrompt) // ← 动态加载
                    .build();

            Message userMsg = Message.builder()
                    .role(Role.USER.getValue())
                    .content(userContent.toString())
                    .build();

            GenerationParam param = GenerationParam.builder()
                    .apiKey("sk-4bd1c10ee2184008a3af580bed54fc3c")
                    .model("qwen-turbo")
                    .messages(Arrays.asList(systemMsg, userMsg))
                    .resultFormat(GenerationParam.ResultFormat.MESSAGE)
                    // 关闭思考过程显示
                    .enableSearch(false)
                    // 禁用提示词缓存
                    .temperature(0.8F)
                    // 设置超时时间
                    .build();

            GenerationResult result = gen.call(param);
            return result.getOutput().getChoices().get(0).getMessage().getContent();

        } catch (Exception e) {
            System.err.println("调用AI接口失败: " + e.getMessage());
            e.printStackTrace();
            return "{}";
        }
    }



    /**
     * 保存分析结果到 MySQL（含 task_id）
     * 优化版本 - 减少数据库交互次数
     */
    @DataSource(DataSourceType.MASTER)
    private void saveAnalysisResultToMySQL(String sessionId, int messageCount, String analysisResult, Long taskId) {
        // 临时调试：打印当前数据源
        System.out.println("【DEBUG】当前数据源: " + DataSourceUtil.getCurrentDataSource());
        try {
            String jsonOnly = extractJsonFromText(analysisResult);
            JsonObject root = JsonParser.parseString(jsonOnly).getAsJsonObject();
            JsonObject sessionStats = root.getAsJsonObject("session_statistics");
            JsonArray transactions = root.getAsJsonArray("transaction_records");

            // 保存会话统计
            ReliableAnalyzedSession session = new ReliableAnalyzedSession();
            session.setSessionId(sessionId);
            session.setTaskId(taskId); // ✅ 保存 task_id
            session.setDialogStartTime(parseDateTime(sessionStats.get("dialog_start_time").getAsString()));
            session.setDialogEndTime(parseDateTime(sessionStats.get("dialog_end_time").getAsString()));
            session.setTotalMessagesCount(sessionStats.get("total_messages_count").getAsInt());
            session.setClientMessageCount(sessionStats.get("client_message_count").getAsInt());
            session.setAgentMessageCount(sessionStats.get("agent_message_count").getAsInt());
            session.setTotalTransactionsDetected(sessionStats.get("total_transactions_detected").getAsInt());
            session.setSuspectAccountCount(sessionStats.get("suspect_account_count").getAsInt());
            session.setContainsSensitiveKeywords(sessionStats.get("contains_sensitive_keywords").getAsBoolean());
            session.setRiskLevel(sessionStats.get("risk_level").getAsString());
            session.setTotalRechargeAttempts(sessionStats.get("total_recharge_attempts").getAsInt());
            session.setSuccessfulRechargeCount(sessionStats.get("successful_recharge_count").getAsInt());
            session.setFailedRechargeCount(sessionStats.get("failed_recharge_count").getAsInt());
            session.setDistinctRechargeMethods(sessionStats.get("distinct_recharge_methods").getAsInt());
            session.setFraudSummary(sessionStats.get("fraud_summary").getAsString());

            // 使用insert方法替代saveOrUpdate避免查询操作导致的数据源问题
            analyzedSessionService.saveOrUpdate(session);

            // 批量保存交易记录
            if (transactions != null && !transactions.isEmpty()) {
                List<ReliableTransactionRecord> records = new ArrayList<>();
                for (JsonElement txElem : transactions) {
                    JsonObject tx = txElem.getAsJsonObject();
                    ReliableTransactionRecord record = new ReliableTransactionRecord();
                    record.setTransactionId(tx.get("transaction_id").getAsString());
                    record.setSessionId(sessionId);
                    record.setTaskId(taskId); // ✅ 保存 task_id

                    JsonObject accountInfo = tx.getAsJsonObject("account_info");
                    record.setAccountType(accountInfo.get("account_type").getAsString());
                    record.setPlatformName(getAsStringOrNull(accountInfo, "platform_name"));
                    record.setAccountCategory(accountInfo.get("account_category").getAsString());

                    record.setRechargeMethodSummary(tx.get("recharge_method_summary").getAsString());
                    record.setAccountIdentifier(getAsStringOrNull(tx.getAsJsonObject("payment_details"), "account_identifier"));
                    record.setQrCodeUrls(getJsonArrayAsStringList(tx.getAsJsonObject("payment_details"), "qr_code_urls"));
                    record.setPaymentPlatformName(getAsStringOrNull(tx.getAsJsonObject("payment_details"), "platform_name"));
                    record.setAmountOrQuantity(getAsBigDecimalOrNull(tx.getAsJsonObject("payment_details"), "amount_or_quantity"));
                    record.setPaymentDescription(getAsStringOrNull(tx.getAsJsonObject("payment_details"), "description"));

                    JsonObject behavior = tx.getAsJsonObject("transaction_behavior");
                    record.setIsSuccess(behavior.get("is_success").getAsBoolean());
                    record.setHasReceiptProof(behavior.get("has_receipt_proof").getAsBoolean());
                    record.setProofImageUrls(getJsonArrayAsStringList(behavior, "proof_image_urls"));

                    record.setTransactionSummary(tx.get("transaction_summary").getAsString());
                    record.setRelatedOriginalSentences(getJsonArrayAsStringList(tx, "related_original_sentences"));

                    // 设置新增字段
                    record.setVictimName(getAsStringOrNull(tx, "victim_name"));
                    record.setIdNumber(getAsStringOrNull(tx, "id_number"));

                    // 设置 delivery_address 中的字段
                    if (tx.has("delivery_address") && !tx.get("delivery_address").isJsonNull()) {
                        JsonObject deliveryAddress = tx.getAsJsonObject("delivery_address");
                        record.setRecipientName(getAsStringOrNull(deliveryAddress, "recipient_name"));
                        record.setFullAddress(getAsStringOrNull(deliveryAddress, "full_address"));
                        record.setContactPhone(getAsStringOrNull(deliveryAddress, "contact_phone"));
                        record.setDeliveryMethod(getAsStringOrNull(deliveryAddress, "delivery_method"));
                    }

                    records.add(record);
                }
                // 使用批量保存方法减少数据库交互次数
                transactionRecordService.saveOrUpdateBatch(records);
            }

            System.out.println("✅ 会话 " + sessionId + " 分析结果已存入 MySQL");

        } catch (Exception e) {
            System.err.println("保存分析结果到 MySQL 失败: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private LocalDateTime parseDateTime(String dateTimeStr) {
        if (dateTimeStr == null || dateTimeStr.isEmpty()) {
            throw new IllegalArgumentException("DateTime string cannot be null or empty");
        }

        // 移除末尾的 .0（例如：2025-09-25 09:07:55.0 → 2025-09-25 09:07:55）
        if (dateTimeStr.endsWith(".0")) {
            dateTimeStr = dateTimeStr.substring(0, dateTimeStr.length() - 2);
        }

        // 尝试多种格式：带偏移量和不带偏移量
        // 支持格式示例：
        // - "2025-09-25 09:07:55"
        // - "2025-09-25 09:07:55+08"
        // - "2025-09-25 09:07:55+0800"
        // - "2025-09-25 09:07:55+08:00"

        // 先尝试带偏移量的格式
        try {
            DateTimeFormatter formatterWithOffset = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ssX");
            OffsetDateTime odt = OffsetDateTime.parse(dateTimeStr, formatterWithOffset);
            return odt.toLocalDateTime();
        } catch (DateTimeParseException e) {
            // 忽略，继续尝试无偏移格式
        }

        // 再尝试不带偏移量的标准格式
        try {
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            return LocalDateTime.parse(dateTimeStr, formatter);
        } catch (DateTimeParseException e) {
            throw new DateTimeParseException("Unable to parse datetime string: " + dateTimeStr, dateTimeStr, 0);
        }
    }

    private String getAsStringOrNull(JsonObject obj, String key) {
        return obj != null && obj.has(key) && !obj.get(key).isJsonNull() ?
                obj.get(key).getAsString() : null;
    }

    private BigDecimal getAsBigDecimalOrNull(JsonObject obj, String key) {
        if (obj != null && obj.has(key) && !obj.get(key).isJsonNull()) {
            JsonElement el = obj.get(key);
            if (el.isJsonPrimitive()) {
                return el.getAsBigDecimal();
            }
        }
        return null;
    }

    private List<String> getJsonArrayAsStringList(JsonObject obj, String key) {
        if (obj != null && obj.has(key) && !obj.get(key).isJsonNull()) {
            JsonArray arr = obj.getAsJsonArray(key);
            List<String> list = new ArrayList<>();
            for (JsonElement el : arr) {
                if (!el.isJsonNull()) {
                    list.add(el.getAsString());
                }
            }
            return list;
        }
        return Collections.emptyList();
    }


    private String extractJsonFromText(String text) {
        int start = text.indexOf('{');
        if (start == -1) {
            throw new IllegalArgumentException("未找到 JSON 起始符号 '{'");
        }
        int end = -1;
        int depth = 0;
        for (int i = start; i < text.length(); i++) {
            char c = text.charAt(i);
            if (c == '{') depth++;
            else if (c == '}') depth--;
            if (depth == 0) {
                end = i + 1;
                break;
            }
        }
        if (end == -1) {
            throw new IllegalArgumentException("JSON 未正确闭合");
        }
        return text.substring(start, end);
    }

    /**
     * 获取PG数据库总量概览数据
     *
     * @return 总量概览数据
     */
    @Override
    public TotalOverviewData getTotalOverviewData() {
        DataSourceUtil.setDataSourceType(DataSourceType.PG.name());
        try {
            String sql = "SELECT " +
                    "(SELECT COUNT(*) FROM chat_messages) AS total_messages, " +
                    "(SELECT total_sessions FROM mv_total_sessions) AS total_sessions, " +
                    "(SELECT COUNT(*) FROM analyzed_sessions) AS analyzed_sessions";

            return jdbcTemplate.queryForObject(sql, (rs, rowNum) -> {
                TotalOverviewData data = new TotalOverviewData();
                data.setTotalMessages(rs.getInt("total_messages"));
                data.setTotalSessions(rs.getInt("total_sessions"));
                data.setAnalyzedSessions(rs.getInt("analyzed_sessions"));
                return data;
            });
        } finally {
            DataSourceUtil.clearDataSourceType();
        }
    }


    /**
     * 首页查询接口，通过seesionid和关键字like查询,这两个有其中一个就行（根据我前端用户选择的条件，比如用户选择关键字，那么他的值就是关键字，我就需要根据关键字去查询，并且要分页）
     *
     * @param sessionId 会话ID（可选）
     * @param keyword 关键字（可选）
     * @param pageNum 页码（仅在使用关键字查询时需要）
     * @param pageSize 每页大小（仅在使用关键字查询时需要）
     * @return 查询结果
     */
    @Override
    @DataSource(value = DataSourceType.PG)
    public List<ChatMessage> findForHomePage(String sessionId, String keyword, Integer pageNum, Integer pageSize) {
        DataSourceUtil.setDataSourceType(DataSourceType.PG.name());
        try {
            StringBuilder sqlBuilder = new StringBuilder();
            sqlBuilder.append("SELECT id, session_id, role, content, created_at, source_type, task_id FROM public.chat_messages WHERE 1=1 ");

            // 如果提供了sessionId，则按sessionId查询，不分页
            if (sessionId != null && !sessionId.isEmpty()) {
                sqlBuilder.append("AND session_id = ? ");
                sqlBuilder.append("ORDER BY created_at DESC");

                String sql = sqlBuilder.toString();
                return jdbcTemplate.query(sql, (rs, rowNum) -> {
                    ChatMessage message = new ChatMessage();
                    message.setId(rs.getLong("id"));
                    message.setSessionId(rs.getString("session_id"));
                    message.setRole(rs.getString("role"));
                    message.setContent(rs.getString("content"));
                    message.setCreatedAt(rs.getTimestamp("created_at").toLocalDateTime());
                    message.setSourceType(rs.getString("source_type"));
                    message.setTaskId(rs.getLong("task_id"));
                    return message;
                }, sessionId);
            }

            // 如果提供了关键字，则按关键字模糊查询content字段，并分页
            if (keyword != null && !keyword.isEmpty()) {
                sqlBuilder.append("AND content ILIKE ? ");
                sqlBuilder.append("ORDER BY created_at DESC ");

                // 分页处理
                if (pageNum != null && pageSize != null) {
                    int offset = (pageNum - 1) * pageSize;
                    sqlBuilder.append("LIMIT ? OFFSET ? ");
                }

                String sql = sqlBuilder.toString();
                return jdbcTemplate.query(sql, (rs, rowNum) -> {
                    ChatMessage message = new ChatMessage();
                    message.setId(rs.getLong("id"));
                    message.setSessionId(rs.getString("session_id"));
                    message.setRole(rs.getString("role"));
                    message.setContent(rs.getString("content"));
                    message.setCreatedAt(rs.getTimestamp("created_at").toLocalDateTime());
                    message.setSourceType(rs.getString("source_type"));
                    message.setTaskId(rs.getLong("task_id"));
                    return message;
                }, prepareKeywordParams(keyword, pageNum, pageSize));
            }

            // 不应该到达这里，因为sessionId和keyword至少有一个不为空
            return new ArrayList<>();
        } finally {
            DataSourceUtil.clearDataSourceType();
        }
    }

    /**
     * 首页分页查询接口，返回查询结果和总数量
     *
     * @param sessionId 会话ID（可选）
     * @param keyword 关键字（可选）
     * @param pageNum 页码
     * @param pageSize 每页大小
     * @return 包含聊天记录列表和总数量的分页结果
     */
    @Override
    @DataSource(value = DataSourceType.PG)
    public ChatMessagePageDTO findForHomePageWithTotal(String sessionId, String keyword, Integer pageNum, Integer pageSize) {
        DataSourceUtil.setDataSourceType(DataSourceType.PG.name());
        try {
            // 构建查询数据的SQL
            StringBuilder dataSqlBuilder = new StringBuilder();
            dataSqlBuilder.append("SELECT session_id, role, content, created_at, task_id FROM public.chat_messages WHERE 1=1 ");

            // 构建统计总数的SQL
            StringBuilder countSqlBuilder = new StringBuilder();
            countSqlBuilder.append("SELECT COUNT(*) FROM public.chat_messages WHERE 1=1 ");

            // 如果提供了sessionId，则按sessionId查询
            if (sessionId != null && !sessionId.isEmpty()) {
                dataSqlBuilder.append("AND session_id = ? ");
                countSqlBuilder.append("AND session_id = ? ");
            }

            // 如果提供了关键字，则按关键字模糊查询content字段
            if (keyword != null && !keyword.isEmpty()) {
                dataSqlBuilder.append("AND content ILIKE ? ");
                countSqlBuilder.append("AND content ILIKE ? ");
            }

            // 数据查询按创建时间倒序排列
            dataSqlBuilder.append("ORDER BY created_at DESC ");

            // 分页处理
            Object[] countParams = new Object[0];
            Object[] dataParams = new Object[0];

            if (sessionId != null && !sessionId.isEmpty() && (keyword == null || keyword.isEmpty())) {
                // 仅sessionId查询，分页
                countParams = new Object[]{sessionId};
                if (pageNum != null && pageSize != null) {
                    int offset = (pageNum - 1) * pageSize;
                    dataSqlBuilder.append("LIMIT ? OFFSET ? ");
                    dataParams = new Object[]{sessionId, pageSize, offset};
                } else {
                    dataParams = new Object[]{sessionId};
                }
            } else if ((sessionId == null || sessionId.isEmpty()) && keyword != null && !keyword.isEmpty()) {
                // 仅关键字查询，分页
                countParams = new Object[]{"%" + keyword + "%"};
                if (pageNum != null && pageSize != null) {
                    int offset = (pageNum - 1) * pageSize;
                    dataSqlBuilder.append("LIMIT ? OFFSET ? ");
                    dataParams = new Object[]{"%" + keyword + "%", pageSize, offset};
                } else {
                    dataParams = new Object[]{"%" + keyword + "%"};
                }
            } else if (sessionId != null && !sessionId.isEmpty() && keyword != null && !keyword.isEmpty()) {
                // sessionId和关键字都查询，分页
                countParams = new Object[]{sessionId, "%" + keyword + "%"};
                if (pageNum != null && pageSize != null) {
                    int offset = (pageNum - 1) * pageSize;
                    dataSqlBuilder.append("LIMIT ? OFFSET ? ");
                    dataParams = new Object[]{sessionId, "%" + keyword + "%", pageSize, offset};
                } else {
                    dataParams = new Object[]{sessionId, "%" + keyword + "%"};
                }
            } else {
                // 无条件查询，分页
                countParams = new Object[]{};
                if (pageNum != null && pageSize != null) {
                    int offset = (pageNum - 1) * pageSize;
                    dataSqlBuilder.append("LIMIT ? OFFSET ? ");
                    dataParams = new Object[]{pageSize, offset};
                }
            }

            String dataSql = dataSqlBuilder.toString();
            String countSql = countSqlBuilder.toString();

            // 查询数据
            List<ChatMessageDTO> records = jdbcTemplate.query(dataSql, (rs, rowNum) -> {
                ChatMessageDTO message = new ChatMessageDTO();
                message.setSessionId(rs.getString("session_id"));
                message.setRole(rs.getString("role"));
                message.setContent(rs.getString("content"));
                message.setCreatedAt(rs.getTimestamp("created_at").toLocalDateTime());
                message.setTaskId(rs.getLong("task_id"));
                return message;
            }, dataParams);

            // 查询总数
            Long total = jdbcTemplate.queryForObject(countSql, Long.class, countParams);

            return new ChatMessagePageDTO(records, total, pageNum, pageSize);
        } finally {
            DataSourceUtil.clearDataSourceType();
        }
    }

    /**
     * 通过sessionId查询聊天记录，只返回role、content和时间字段
     *
     * @param sessionId 会话ID
     * @return 聊天记录列表
     */
    @Override
    @DataSource(value = DataSourceType.PG)
    public List<ChatMessageSessionDTO> findSessionMessages(String sessionId, Integer pageNum, Integer pageSize) {
        DataSourceUtil.setDataSourceType(DataSourceType.PG.name());
        try {
            // 安全处理分页参数（防止 null 或非法值）
            int page = (pageNum == null || pageNum < 1) ? 1 : pageNum;
            int size = (pageSize == null || pageSize < 1) ? 10 : pageSize;

            // 计算 OFFSET（PostgreSQL 从 0 开始）
            int offset = (page - 1) * size;

            // PostgreSQL 分页 SQL：LIMIT + OFFSET
            String sql = "SELECT role, content, created_at " +
                    "FROM public.chat_messages " +
                    "WHERE session_id = ? " +
                    "ORDER BY created_at ASC " +
                    "LIMIT ? OFFSET ?";

            return jdbcTemplate.query(sql, (rs, rowNum) -> {
                ChatMessageSessionDTO message = new ChatMessageSessionDTO();
                message.setRole(rs.getString("role"));
                message.setContent(rs.getString("content"));
                message.setCreatedAt(rs.getTimestamp("created_at").toLocalDateTime());
                return message;
            }, sessionId, size, offset);
        } finally {
            DataSourceUtil.clearDataSourceType();
        }
    }

    /**
     * 准备关键字查询参数
     * @param keyword 关键字
     * @param pageNum 页码
     * @param pageSize 每页大小
     * @return 参数数组
     */
    private Object[] prepareKeywordParams(String keyword, Integer pageNum, Integer pageSize) {
        List<Object> params = new ArrayList<>();
        params.add("%" + keyword + "%");

        if (pageNum != null && pageSize != null) {
            params.add(pageSize);
            params.add((pageNum - 1) * pageSize);
        }

        return params.toArray();
    }

    /**
     * 查询具有唯一起始和目标页面的会话记录
     *
     * @param referrerUrl 来源页URL（可选）
     * @param pageUrl 对话页URL（可选）
     * @param pageNum 页码
     * @param pageSize 每页大小
     * @return 查询结果列表
     */
    @DataSource(value = DataSourceType.PG)
    public List<SessionPageTransitionDTO> findUniqueSessionPageTransitions(
            String referrerUrl, String pageUrl, Integer pageNum, Integer pageSize) {

        DataSourceUtil.setDataSourceType(DataSourceType.PG.name());
        try {
            int limit = (pageSize == null || pageSize < 1) ? 10 : pageSize;
            int offset = (pageNum == null || pageNum < 1) ? 0 : (pageNum - 1) * limit;

            String sql = "  SELECT session_id, created_at, referrer_url, page_url\n" +
                    "            FROM (\n" +
                    "                SELECT \n" +
                    "                    session_id, created_at, referrer_url, page_url,\n" +
                    "                    ROW_NUMBER() OVER (\n" +
                    "                        PARTITION BY session_id, referrer_url, page_url \n" +
                    "                        ORDER BY created_at ASC\n" +
                    "                    ) AS rn\n" +
                    "                FROM chat_messages\n" +
                    "                WHERE referrer_url IS NOT NULL \n" +
                    "                  AND page_url IS NOT NULL \n" +
                    "                  AND created_at >= NOW() - INTERVAL '30 days'\n" +
                    "                  AND (? = '' OR referrer_url ILIKE ?)\n" +
                    "                  AND (? = '' OR page_url ILIKE ?)\n" +
                    "            ) t\n" +
                    "            WHERE rn = 1\n" +
                    "            ORDER BY created_at DESC\n" +
                    "            LIMIT ? OFFSET ?";

            String referrerParam = (referrerUrl != null && !referrerUrl.isEmpty())
                    ? "%" + referrerUrl + "%" : "";
            String pageParam = (pageUrl != null && !pageUrl.isEmpty())
                    ? "%" + pageUrl + "%" : "";

            return jdbcTemplate.query(sql, (rs, rowNum) -> {
                SessionPageTransitionDTO dto = new SessionPageTransitionDTO();
                dto.setSessionId(rs.getString("session_id"));
                dto.setCreatedAt(rs.getTimestamp("created_at").toLocalDateTime());
                dto.setReferrerUrl(rs.getString("referrer_url"));
                dto.setPageUrl(rs.getString("page_url"));
                return dto;
            }, referrerParam, referrerParam, pageParam, pageParam, limit, offset);
        } finally {
            DataSourceUtil.clearDataSourceType();
        }
    }

    @Override
    @DataSource(value = DataSourceType.PG)
    public SessionPageTransitionPageDTO findUniqueSessionPageTransitionsWithTotal(
            String referrerUrl, String pageUrl, Integer pageNum, Integer pageSize) {

        DataSourceUtil.setDataSourceType(DataSourceType.PG.name());
        try {
            // 先查询总数
            String countSql = "SELECT COUNT(*) FROM (" +
                    "SELECT 1 FROM (" +
                    "    SELECT session_id, referrer_url, page_url " +
                    "    FROM (" +
                    "        SELECT session_id, referrer_url, page_url, " +
                    "               ROW_NUMBER() OVER (PARTITION BY session_id, referrer_url, page_url ORDER BY created_at ASC) AS rn " +
                    "        FROM chat_messages " +
                    "        WHERE referrer_url IS NOT NULL AND page_url IS NOT NULL " +
                    "          AND created_at >= NOW() - INTERVAL '30 days'" +
                    "          AND (? = '' OR referrer_url ILIKE ?)" +
                    "          AND (? = '' OR page_url ILIKE ?)" +
                    "    ) t " +
                    "    WHERE rn = 1" +
                    ") subquery) AS count_table";

            String referrerParam = (referrerUrl != null && !referrerUrl.isEmpty())
                    ? "%" + referrerUrl + "%" : "";
            String pageParam = (pageUrl != null && !pageUrl.isEmpty())
                    ? "%" + pageUrl + "%" : "";

            Long total = jdbcTemplate.queryForObject(countSql, Long.class,
                    referrerParam, referrerParam, pageParam, pageParam);

            // 再查询具体数据
            List<SessionPageTransitionDTO> records = findUniqueSessionPageTransitions(referrerUrl, pageUrl, pageNum, pageSize);

            return new SessionPageTransitionPageDTO(records, total, pageNum, pageSize);
        } finally {
            DataSourceUtil.clearDataSourceType();
        }
    }
    /**
     * 获取指定时间后新增的聊天消息（用于 IP 分析）
     */
    @Override
    @DataSource(DataSourceType.PG)
    public List<ChatMessageRowForIp> getNewMessagesForIpAnalysis(LocalDateTime sinceTime) {
        DataSourceUtil.setDataSourceType(DataSourceType.PG.name());
        try {
            String sql = sinceTime == null
                    ? "SELECT session_id, visitor_ip, created_at FROM chat_messages WHERE visitor_ip IS NOT NULL"
                    : "SELECT session_id, visitor_ip, created_at FROM chat_messages WHERE visitor_ip IS NOT NULL AND created_at > ?";

            if (sinceTime == null) {
                return jdbcTemplate.query(sql, (rs, rowNum) -> {
                    ChatMessageRowForIp r = new ChatMessageRowForIp();
                    r.setSessionId(rs.getString("session_id"));
                    r.setVisitorIp(rs.getString("visitor_ip"));
                    Timestamp timestamp = rs.getTimestamp("created_at");
                    r.setCreatedAt(timestamp);
                    return r;
                });
            } else {
                return jdbcTemplate.query(sql, (rs, rowNum) -> {
                    ChatMessageRowForIp r = new ChatMessageRowForIp();
                    r.setSessionId(rs.getString("session_id"));
                    r.setVisitorIp(rs.getString("visitor_ip"));
                    Timestamp timestamp = rs.getTimestamp("created_at");
                    r.setCreatedAt(timestamp);
                    return r;
                }, Timestamp.valueOf(sinceTime));
            }
        } finally {
            DataSourceUtil.clearDataSourceType();
        }
    }



    // 新增方法：供分片合并后调用
    @SneakyThrows
    public void importExcelFromFile(File file, String originalFilename, Long taskId) {
        ReliableUserLite currentUser = JwtUtils.getCurrentUser();
        log.info("开始处理Excel文件（来自分片合并），任务ID: {}", taskId);

        ReliabUploadRecord uploadRecord = new ReliabUploadRecord();
        uploadRecord.setTaskId(taskId);
        uploadRecord.setFileName(originalFilename); // ← 使用传入的原始文件名
        uploadRecord.setStatus(UploadRecordStatus.PROCESSING.getCode());
        uploadRecordService.save(uploadRecord);


        LambdaQueryWrapper<ReliableTemplateAnalysisTask> reliableTask = new LambdaQueryWrapper<>();
        reliableTask.eq(ReliableTemplateAnalysisTask::getId, taskId);

        ReliableTemplateAnalysisTask task = templateAnalysisTaskService.getOne(reliableTask);
        task.setUploadRecordId(uploadRecord.getId());
        templateAnalysisTaskService.updateById(task);


        // 关键：调用新的 async 方法，接受 File
        importExcelAsyncFromFile(file, taskId, uploadRecord.getId()).get();
    }


    @Async
    public CompletableFuture<Void> importExcelAsyncFromFile(File file, Long taskId, Long uploadId) {
        // 逻辑几乎和 importExcelAsync 一样，只是不再需要 file.transferTo()
        long startTime = System.currentTimeMillis();
        log.info("开始处理Excel文件（File 版本），任务ID: {}, 上传ID: {}", taskId, uploadId);

        ChatMessageImportListener listener = null;
        try {
            // ✅ 直接使用 file，不再创建 tempFile
            DynamicDataSourceContextHolder.setDataSourceType(DataSourceType.PG.name());
            listener = new ChatMessageImportListener(taskId, uploadId, this);
            // 忽略超链接以避免URI格式异常
            EasyExcel.read(file, ChatMessageExcelDTO.class, listener)
                    .ignoreEmptyRow(Boolean.TRUE)
                    .autoCloseStream(Boolean.TRUE)
                    .sheet()
                    .doRead();

            DynamicDataSourceContextHolder.setDataSourceType(DataSourceType.MASTER.name());
            uploadRecordService.updateUploadRecordSuccess(
                    uploadId,
                    listener.getTotalRows(),
                    listener.getSuccessRows(),
                    listener.getFailedRows()
            );
            long endTime = System.currentTimeMillis();
            log.info("✅ Excel导入完成（File 版本），总耗时: {}ms", (endTime - startTime));
            return CompletableFuture.completedFuture(null);
        } catch (Exception e) {
            // 异常处理逻辑完全复用（略）
            log.error("❌ Excel导入过程中发生异常", e);
            try {
                DynamicDataSourceContextHolder.setDataSourceType(DataSourceType.MASTER.name());
                String errorMsg = e.getMessage() != null ? e.getMessage() : "未知错误";
                uploadRecordService.updateUploadRecordFailed(uploadId, errorMsg);
            } catch (Exception ex) {
                log.error("⚠️ 更新上传记录失败（异常处理中）", ex);
            }
            throw new RuntimeException("Excel导入失败", e);
        } finally {
            DynamicDataSourceContextHolder.clearDataSourceType();
            // 注意：不再删除 tempFile，因为 file 是合并后的临时文件，由调用方负责清理
        }
    }



    /**
     * 根据任务ID统计独立会话数
     *
     * @param taskId 任务ID
     * @return 独立会话数
     */
    @Override
    @DataSource(value = DataSourceType.PG)
    public Long countDistinctSessionsByTaskId(Long taskId) {
        DataSourceUtil.setDataSourceType(DataSourceType.PG.name());
        try {
            String sql = "SELECT COUNT(DISTINCT session_id) FROM public.chat_messages WHERE task_id = ?";
            return jdbcTemplate.queryForObject(sql, Long.class, taskId);
        } finally {
            DataSourceUtil.clearDataSourceType();
        }
    }








}